﻿#include "ssAsioNetWork.h"
#include "log/ssLogThread.h"
#include "message/ssMsgIdDef.h"
#include "message/ssNetMsgHead.h"
#include "container/uniBind.h"

namespace StarSeeker
{

ssAsioNetWork::ssAsioNetWork(void)
	:m_sessionCount(100),m_threadCount(1),
	m_recvBuffSize(SS_ASIO_PACK_SIZE), m_sendBuffSize(SS_ASIO_PACK_SIZE), m_isClient(false), 
	m_exiting(false), m_checkFullCount(false), m_autoConnectInterval(10), m_netName(" ")
{
	m_sessionID.Init(1, 400000000);
}

ssAsioNetWork::~ssAsioNetWork(void)
{
	StopNetWork();
}

void ssAsioNetWork::InitParam(const char *ip, const u16 port, u32 sessionCount, u32 threadCount, u32 recvBuffSize, u32 sendBuffSize)
{
	m_sessionCount = sessionCount;
	m_threadCount = threadCount;
	m_recvBuffSize = recvBuffSize;
	m_sendBuffSize = sendBuffSize;
	if (ip && port)
	{
		MakeAddress(ip, port, m_listenAddr);
	}
}

bool ssAsioNetWork::StartNetWork(bool isClient)
{
	if (m_threads.empty() && !m_exiting)
	{
		m_isClient = isClient;
		for (u32 i = 0; i < m_threadCount; ++i)
		{
			bThreadPtr pThread(new ssAsioThread());
			m_threads.push_back(pThread);
		}
		if (m_isClient)
		{
			ConnectToServer();
		}
		else
		{
			bool isv4 = m_listenAddr.address().is_v4();
			try
			{
				m_acceptor = SS_MAKE_SHARED<asio::ip::tcp::acceptor>(m_threads[0]->m_service);
				m_acceptor->open(isv4?asio::ip::tcp::v4():asio::ip::tcp::v6());
				m_acceptor->set_option(asio::ip::tcp::acceptor::reuse_address(true));
				m_acceptor->bind(m_listenAddr);
				m_acceptor->listen();
			}
			catch (asio::system_error& e)
			{
				SS_LOG->ErrorEx("%s start listen error:%s", m_netName.c_str(), e.what());
				return false;
			}
			catch (...)
			{
				SS_LOG->Error("ssAsioNetWork::StartNetWork error");
				return false;
			}
			DoAccept();
			SS_LOG->PrintEx("%s start listen %s:%d", m_netName.c_str(), m_listenAddr.address().to_string().c_str(), m_listenAddr.port());
		}
		if (m_exiting)
			return false;
	}
	return false;
}

void ssAsioNetWork::StopNetWork()
{
	if (!m_exiting)
	{
		m_exiting = true;
		if (!m_isClient && m_acceptor)
		{
			//m_acceptor->cancel();
			m_acceptor->close();
		}
		for (size_t i = 0; i < m_threads.size(); ++i)
		{
			if (m_threads[i])
			{
				m_threads[i]->m_service.stop();
				m_threads[i]->m_thread.join();
				m_threads[i]->ClearSession();
				//SS_SAFE_DELETE(m_threads[i])
			}
		}
		m_threads.clear();
	}
}

void ssAsioNetWork::DoAccept()
{
	u32 id = m_sessionID.GetID();
	SS_SHARED_PTR<asio::ip::tcp::socket> sk2(new asio::ip::tcp::socket(m_threads[id%m_threadCount]->m_service));
	m_acceptor->async_accept(*sk2, SS_BIND(&ssAsioNetWork::HandleAccept, this, id, sk2, SS_BIND_HOLDERS::_1));
}

void ssAsioNetWork::HandleAddSession(u32 threadId, ssAsioSessionPtr& ps)
{
	m_threads[threadId]->m_sessionMap.insert(std::pair<u32, ssAsioSessionPtr>(ps->GetID(), ps));
	DoReceive(ps);
	OnNewSession(*ps);
}

void ssAsioNetWork::DoReceive(ssAsioSessionPtr& ptr)
{
	if (ptr->m_recvBuf.ResidueSize()<100)
	{
		SS_LOG->ErrorEx("%s %d recvBuf residueSize<100:%u", m_netName.c_str(), ptr->m_sessionID, (u32)ptr->m_recvBuf.ResidueSize());
		if (m_checkFullCount)
		{
			if (ptr->m_fullCount++ > 50)
			{
				u32 threadId = ptr->m_sessionID % m_threadCount;
				HandleCloseSocket(threadId, ptr->m_sessionID, CLOSE_BY_READBUF);
				return;
			}
		}
	}
	ptr->m_socket->async_read_some(asio::buffer(ptr->m_recvBuf.CurrentBuf(), ptr->m_recvBuf.ResidueSize()), SS_BIND(&ssAsioNetWork::HandleAfterRead, this, SS_BIND_HOLDERS::_1, SS_BIND_HOLDERS::_2, ptr));
}

void ssAsioNetWork::HandleAccept(u32 sessionID, SS_SHARED_PTR<asio::ip::tcp::socket>& s, const asio::error_code& error)
{
	if (s)
	{
		if (!error)
		{
			if (GetSessionCount() > m_sessionCount)
			{
				asio::error_code ec1, ec2;
				s->shutdown(asio::ip::tcp::socket::shutdown_both, ec1);
				s->close(ec2);
			}
			else
			{
				//!!需要对象池
				ssAsioSessionPtr pSession(CreateSession());
				pSession->Init(sessionID, s, m_recvBuffSize, m_sendBuffSize, this);
				u32 threadId = sessionID % m_threadCount;
				m_threads[threadId]->m_service.post(SS_BIND(&ssAsioNetWork::HandleAddSession, this, threadId, pSession));
			}
		}
		else
		{
            std::string errstr("HandleAccept %u");
            errstr +=   error.message();
            SS_LOG->ErrorEx(errstr.c_str(), sessionID);
			asio::error_code ec1, ec2;
			s->shutdown(asio::ip::tcp::socket::shutdown_both, ec1);
			s->close(ec2);
		}
	}
	DoAccept();
}

void ssAsioNetWork::DoCloseSocket(u32 sessionId, eAsioClose errTypr, i32 err)
{
	if (!m_threads.empty())
	{
		u32 threadId = sessionId % m_threadCount;
		m_threads[threadId]->m_service.post(SS_BIND(&ssAsioNetWork::HandleCloseSocket, this, threadId, sessionId, errTypr));
	}
}

void ssAsioNetWork::DoSendData(u32 sessionId, SS_SHARED_ARRAY<char>& buf, u32 len)
{
	if (!m_threads.empty())
	{
		u32 threadId = sessionId % m_threadCount;
		m_threads[threadId]->m_service.post(SS_BIND(&ssAsioNetWork::HandleSendData, this, threadId, sessionId, buf, len));
	}
}

void ssAsioNetWork::DoSendData(u32 sessionId, char* buf, u32 len)
{
	if (!m_threads.empty())
	{
		u32 threadId = sessionId % m_threadCount;
		m_threads[threadId]->m_service.post(SS_BIND(&ssAsioNetWork::HandleSendData, this, threadId, sessionId, make_shared_array<char>(buf), len));
	}
	else
	{
		delete[] buf;
	}
}

void ssAsioNetWork::DoSendDataMulti(std::vector<u32>& sessionId, char* buf, u32 len)
{
	SS_SHARED_ARRAY<char> data = make_shared_array(buf);
	if (!m_threads.empty())
	{
		for (size_t i = 0; i < sessionId.size(); i++)
		{
			u32 threadId = sessionId[i] % m_threadCount;
			m_threads[threadId]->m_service.post(SS_BIND(&ssAsioNetWork::HandleSendData, this, threadId, sessionId[i], data, len));
		}
	}
}

void ssAsioNetWork::DoSendDataAll(char* buf, u32 len)
{
	SS_SHARED_ARRAY<char> data = make_shared_array(buf);
	if (!m_threads.empty())
	{
		for (size_t i = 0; i < m_threads.size(); i++)
			m_threads[i]->m_service.post(SS_BIND(&ssAsioNetWork::HandleSendAll, this, (u32)i, data, len));
	}
}

//StarSeeker::u32 ssAsioNetWork::GetClientID()
//{
//	if (m_isClient)
//	{
//		for (size_t i = 0; i < m_threads.size(); ++i)
//		{
//			if (!m_threads[i]->m_sessionMap.empty())
//			{
//				return m_threads[i]->m_sessionMap.begin()->first;
//			}
//		}
//	}
//	return 0;
//}

StarSeeker::u32 ssAsioNetWork::GetSessionCount()
{
	u32 ct = 0;
	for (size_t i = 0; i < m_threads.size(); ++i)
	{
		ct += m_threads[i]->GetSessionCount();
	}
	return ct;
}

StarSeeker::lint ssAsioNetWork::GetSendEventCount()
{
	lint ct = 0;
	for (size_t i = 0; i < m_threads.size(); ++i)
	{
		ct += m_threads[i]->m_sendEventCount;
	}
	return ct;
}

StarSeeker::lint ssAsioNetWork::GetSendQueueCount()
{
	lint ct = 0;
	for (size_t i = 0; i < m_threads.size(); ++i)
	{
		ct += m_threads[i]->m_sendQueueCount;
	}
	return ct;
}

void ssAsioNetWork::OnSessionMessage(ssAsioSession& session, u32 msgID, char* data)
{
	if (data)
	{
		delete[] data;
	}
}

void ssAsioNetWork::SetThreadName(const char* name, const wchar_t* wname)
{
	for (u32 i = 0; i < m_threadCount; ++i)
	{
		m_threads[i]->SetThreadName(i, name, wname);
	}
}

void ssAsioNetWork::MakeAddress(const char *ip, const u16 port, asio::ip::tcp::endpoint& out)
{
	asio::ip::address addrip;
	addrip = addrip.from_string(ip);
	out.address(addrip);
	out.port(port);
}

void ssAsioNetWork::HandleCloseSocket(u32 threadId, u32 sessionId, eAsioClose origin)
{
	bThreadPtr thread = m_threads.at(threadId);
	ssAsioSessionPtr ps = thread->FindSession(sessionId);
	if (ps)
	{
		try
		{
			asio::error_code ec1, ec2;
			ps->GetSocket()->shutdown(asio::ip::tcp::socket::shutdown_both, ec1);
			ps->GetSocket()->close(ec2);
		}
		catch (asio::system_error& e)
		{
			SS_LOG->ErrorEx("%s HandleCloseSocket %d by%d %s", m_netName.c_str(), sessionId, origin, e.what());
		}
		catch (asio::error_code& e)
		{
			SS_LOG->ErrorEx("%s HandleCloseSocket %d by%d %s", m_netName.c_str(), sessionId, origin, e.message().c_str());
		}
		catch (...)
		{
		}
		thread->m_sendQueueCount -= (lint)ps->m_sendQue.m_queue.size();
		thread->m_sessionMap.erase(sessionId);
		//delete ps;
		OnSessionClose(sessionId);
	}
	//else
	//{
	//	SS_LOG->ErrorEx("%s HandleCloseSocket sessionId not found:%d by%d",  m_netName.c_str(), sessionId, origin);
	//}
}

void ssAsioNetWork::HandleSendData(u32 threadId, u32 sessionId, SS_SHARED_ARRAY<char>& buf, u32 len)
{
	if (buf && len)
	{
		if (len < MSG_MAX_SIZE)
		{
			bThreadPtr thread = m_threads.at(threadId);
			ssAsioSessionPtr ps = thread->FindSession(sessionId);
			if (ps)
			{
				SendOneData(thread, ps, buf, len);
			}
		}
		else
		{
			SS_LOG->ErrorEx("%s HandleSendData data too long:%d", m_netName.c_str(), sessionId);
		}
	}
}

void ssAsioNetWork::HandleSendAll(u32 threadId, SS_SHARED_ARRAY<char>& buf, u32 len)
{
	if (buf && len)
	{
		if (len < MSG_MAX_SIZE)
		{
			bThreadPtr thread = m_threads.at(threadId);
			std::map<u32, ssAsioSessionPtr>& smap = thread->m_sessionMap;
			for (std::map<u32, ssAsioSessionPtr>::iterator i = smap.begin(); i != smap.end(); ++i)
			{
				SendOneData(thread, i->second, buf, len);
			}
		}
		else
		{
			SS_LOG->Error("HandleSendAll data too long");
		}
	}
}

void ssAsioNetWork::HandleAfterRead(const asio::error_code & error, size_t bytes, ssAsioSessionPtr& ps)
{
	u32 sessionId = ps->GetID();
	if (error)
	{
		//std::string errstr("HandleAfterRead %u ");
		//errstr += error.message();
		//SS_LOG->ErrorEx(errstr.c_str(), sessionId);
		HandleCloseSocket(sessionId%m_threadCount, sessionId, CLOSE_BY_READ);
	}
	else
	{
		ssAsioSessionPtr session = m_threads.at(sessionId%m_threadCount)->FindSession(sessionId);
		if (session)
		{
			session->m_recvBuf.AddSize(bytes);
			session->OnReceive();
			DoReceive(session);
		}
	}
}

void ssAsioNetWork::HandleAfterWrite(const asio::error_code & error, size_t bytes, ssAsioSessionPtr& ps, SS_SHARED_ARRAY<char>& buf)
{
	u32 sessionId = ps->GetID();
	bThreadPtr thread = m_threads.at(sessionId%m_threadCount);
	thread->m_sendEventCount--;
	if (error)
	{
		//std::string errstr("HandleAfterWrite %u ");
		//errstr += error.message();
		//SS_LOG->ErrorEx(errstr.c_str(), sessionId);
		HandleCloseSocket(sessionId%m_threadCount, sessionId, CLOSE_BY_WRITE);
	}
	else
	{
		ssAsioSessionPtr session = thread->FindSession(sessionId);
		if (session)
		{
			session->m_sending = false;
			if (!session->m_sendQue.m_queue.empty())//对列不为空则合并队列
				SendSessionQueue(thread, session);
		}
	}
}

bool ssAsioNetWork::IsConnect()
{
	if (m_isClient)
	{
		return GetSessionCount() > 0;
	}
	return true;
}

u32 ssAsioNetWork::ConnectToServer(u32 sessionId)
{
	return ConnectToServer(sessionId, 0, 0);
}

u32 ssAsioNetWork::ConnectToServer(u32 sessionId, const char *ip, const u16 port)
{
	if (GetSessionCount() > 0)
	{
		return 0;
	}
	if (m_threads.empty())
	{
		m_isClient = true;
		for (u32 i = 0; i < m_threadCount; ++i)
		{
			bThreadPtr pThread(new ssAsioThread());
			m_threads.push_back(pThread);
		}
	}
	u32 id = sessionId > 0 ? sessionId : m_sessionID.GetID();
	u32 threadID = id % m_threadCount;
	asio::ip::tcp::endpoint addr;
	if (ip && port)
		MakeAddress(ip, port, addr);
	else
		addr = m_listenAddr;
	if (sessionId == 0)//新连接使用传入地址
	{
		m_connectionAddr[id] = addr;
	}
	else//旧连接优先使用记录的地址
	{
		std::map<u32, asio::ip::tcp::endpoint>::iterator itAddr = m_connectionAddr.find(id);
		if (itAddr == m_connectionAddr.end())
			m_connectionAddr[id] = addr;
		else
			addr = itAddr->second;
	}

	asio::error_code err;
	SS_SHARED_PTR<asio::ip::tcp::socket> sk(new asio::ip::tcp::socket(m_threads[threadID]->m_service));
	do
	{
		SS_LOG->PrintEx("%s connect to %s:%d", m_netName.c_str(), addr.address().to_string().c_str(), addr.port());
		sk->connect(addr, err);
		if (!err)
		{
			//!!需要对象池
			ssAsioSessionPtr pSession(CreateSession());
			pSession->Init(id, sk, m_recvBuffSize, m_sendBuffSize, this);
			m_threads[threadID]->m_sessionMap.insert(std::pair<u32, ssAsioSessionPtr>(id, pSession));
			DoReceive(pSession);
			OnNewSession(*pSession);
			return id;
		}

		ssThreadBase::Sleep((u32)m_autoConnectInterval * 1000);
	} while (err && !m_exiting);
	return 0;
}

void ssAsioNetWork::AasyncConnectToServer(u32 sessionId)
{
	AasyncConnectToServer(sessionId, 0, 0);
}

void ssAsioNetWork::AasyncConnectToServer(u32 sessionId, const char *ip, const u16 port)
{
	if (m_threads.empty())
	{
		m_isClient = true;
		for (u32 i = 0; i < m_threadCount; ++i)
		{
			bThreadPtr pThread(new ssAsioThread());
			m_threads.push_back(pThread);
		}
	}

	u32 id = sessionId > 0 ? sessionId : m_sessionID.GetID();
	u32 threadID = id % m_threadCount;
	asio::ip::tcp::endpoint addr;
	if (ip && port)
		MakeAddress(ip, port, addr);
	else
		addr = m_listenAddr;
	if (sessionId == 0)//新连接使用传入地址
	{
		m_connectionAddr[id] = addr;
	}
	else//旧连接优先使用记录的地址
	{
		std::map<u32, asio::ip::tcp::endpoint>::iterator itAddr = m_connectionAddr.find(id);
		if (itAddr == m_connectionAddr.end())
			m_connectionAddr[id] = addr;
	}

	SS_SHARED_PTR<asio::ip::tcp::socket> sk(new asio::ip::tcp::socket(m_threads[threadID]->m_service));
	SS_SHARED_PTR<asio::steady_timer> t(new asio::steady_timer(m_threads[threadID]->m_service, std::chrono::seconds(m_autoConnectInterval)));
	t->async_wait(std::bind(&ssAsioNetWork::HandleConnectToServer, this, std::placeholders::_1, id, sk, t));
}

void ssAsioNetWork::HandleConnectToServer(const asio::error_code& error, u32 sessionID, SS_SHARED_PTR<asio::ip::tcp::socket>& s, SS_SHARED_PTR<asio::steady_timer>& timer)
{
	if (error) return;

	asio::ip::tcp::endpoint addr;
	std::map<u32, asio::ip::tcp::endpoint>::iterator itAddr = m_connectionAddr.find(sessionID);
	if (itAddr != m_connectionAddr.end())
		addr = itAddr->second;
	else
		return;

	SS_LOG->PrintEx("%s try connect to %s:%d", m_netName.c_str(), addr.address().to_string().c_str(), addr.port());
	u32 threadID = sessionID % m_threadCount;
	asio::error_code connError;
	s->connect(addr, connError);
	if (!connError)
	{
		//!!需要对象池
		ssAsioSessionPtr pSession(CreateSession());
		pSession->Init(sessionID, s, m_recvBuffSize, m_sendBuffSize, this);
		m_threads[threadID]->m_sessionMap.insert(std::pair<u32, ssAsioSessionPtr>(sessionID, pSession));
		DoReceive(pSession);
		OnNewSession(*pSession);
	}
	else
	{
		if (!m_exiting)
		{
			timer->expires_from_now(std::chrono::seconds(m_autoConnectInterval));
			timer->async_wait(std::bind(&ssAsioNetWork::HandleConnectToServer, this, std::placeholders::_1, sessionID, s, timer));
		}
	}
}

void ssAsioNetWork::SendOneData(bThreadPtr& pt, ssAsioSessionPtr& ps, SS_SHARED_ARRAY<char>& buf, u32 len)
{
	//正在发送则加入队列
	if (ps->m_sending)
	{
		ps->m_sendQue.Push(buf, len);
		pt->m_sendQueueCount++;
	}
	else if (!ps->m_sendQue.m_queue.empty())//对列不为空则合并队列
	{
		ps->m_sendQue.Push(buf, len);
		pt->m_sendQueueCount++;
		SendSessionQueue(pt, ps);
	}
	else//对列为空，直接发送
	{
		pt->m_sendEventCount++;
		ps->m_sending = true;
		WriteMsgIntoSendBuffer(buf, len);
		asio::async_write(*ps->m_socket, asio::buffer(buf.get(), len),
			SS_BIND(&ssAsioNetWork::HandleAfterWrite, this, SS_BIND_HOLDERS::_1, SS_BIND_HOLDERS::_2, ps, buf));
	}
}

void ssAsioNetWork::SendSessionQueue(bThreadPtr& pt, ssAsioSessionPtr& ps)
{
	SS_SHARED_ARRAY<char> sendbuf = make_shared_array(new char[ps->m_sendQue.m_totalSize]);
	ssWriteBytes wb; wb.Set(sendbuf.get(), ps->m_sendQue.m_totalSize);
	pt->m_sendEventCount++;
	pt->m_sendQueueCount -= ps->m_sendQue.m_queue.size();
	ps->m_sending = true;
	for (auto itr: ps->m_sendQue.m_queue)
	{
		WriteMsgIntoSendBuffer(wb, itr.first.get(), itr.second);
	}
	asio::async_write(*ps->m_socket, asio::buffer(wb.Buf(), wb.Size()),
		SS_BIND(&ssAsioNetWork::HandleAfterWrite, this, SS_BIND_HOLDERS::_1, SS_BIND_HOLDERS::_2, ps, sendbuf));
	ps->m_sendQue.Clear();

	if (wb.Size() > SS_1MB)
		SS_LOG->PrintEx("%s session%u sendBuff more than 1mb size:%lldkb", m_netName.c_str(), ps->GetID(), wb.Size()/1024);
}

ssAsioSession * ssAsioNetWork::CreateSession()
{
	return new ssAsioSession();
}

void ssAsioNetWork::WriteMsgIntoSendBuffer(SS_SHARED_ARRAY<char>& buf, u32 & len)
{
}

void ssAsioNetWork::WriteMsgIntoSendBuffer(ssWriteBytes& wb, const char* data, u32 len)
{
	wb.WriteData(data, len);
}

}
